package ru.spbau.database.group3.filesAndAccess;

import ru.spbau.database.group3.engine.Column;
import ru.spbau.database.group3.engine.Condition;
import ru.spbau.database.group3.memoryManager.BufferManager;

import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.*;

/**
 * @author: stas
 * @date: 2/2/13
 * @time: 9:36 PM
 */
public class Page implements Serializable {

    public Page(int pid) {
        this.pid = pid;
    }

    public int getId() {
        return pid;
    }

//    public Collection<? extends Record> getRecords(Collection<Condition> c) throws IOException {
//    	Set<Record> records = new HashSet<Record>();
//    	for (Condition condition : c) {
//    		records.addAll(getRecords(condition));
//    	}
//    	return records;
//    }
    
    public List<Record> getRecords(List<Condition> conditions) throws IOException {
        if(conditions.isEmpty()) {
            return getAllRecords();
        }
        LinkedList<Record> records = new LinkedList<Record>();
        BitSet slotMap = readSlotMap(bufferManager.getPageData(pid));
        int i = slotMap.nextSetBit(0);
        while(i >= 0) {
            Record r = new Record(pid, i);
            if(checkAllConditions(r, conditions)) {
                records.add(r);
            }
            i = slotMap.nextSetBit(i + 1);
        }
        return records;
    }

    public List<Record> deleteRecords(List<Condition> conditions) throws IOException {
        LinkedList<Record> records = new LinkedList<Record>();
        BitSet slotMap = readSlotMap(bufferManager.getPageData(pid));
        int i = slotMap.nextSetBit(0);
        while(i >= 0) {
            Record r = new Record(pid, i);
            if(checkAllConditions(r, conditions)) {
                records.add(r);
                slotMap.clear(i);
            }
            i = slotMap.nextSetBit(i + 1);
        }
        writeSlotMap(slotMap);
        return records;
    }




    public Record insertRecord(ByteBuffer recordData) throws IOException {
        recordData.position(0);
        ByteBuffer pageData = bufferManager.getPageData(pid);
        BitSet slotMap = readSlotMap(pageData);
        int i = slotMap.nextClearBit(0);
        slotMap.set(i);
        pageData.position(SLOT_MAP_POSITION);
        pageData.put(slotMap.toByteArray());
        int recordPosition = getDataIndex() + i * getSlotSize();
        pageData.position(recordPosition);
        pageData.put(recordData);
        pageData.position(0);
        return new Record(pid, i);
    }

    public boolean isFull() throws IOException {
        ByteBuffer pageData = bufferManager.getPageData(pid);
        BitSet slotMap = readSlotMap(pageData);
        int slotMapSize = pageData.getInt(SLOT_MAP_SIZE_POSITION);
        int card = slotMap.cardinality();
        return slotMapSize * 8 == card;
    }


    public int getSlotSize() throws IOException {
        return bufferManager.getPageData(pid).getInt(RECORD_SIZE_POSITION);
    }

    public List<Record> getAllRecords() throws IOException {
        ArrayList<Record> records = new ArrayList<Record>();
        BitSet slotMap = readSlotMap(bufferManager.getPageData(pid));
        int i = slotMap.nextSetBit(0);
        while(i >= 0) {
            records.add(new Record(pid, i));
            i = slotMap.nextSetBit(i + 1);
        }
        return records;
    }

    public void clearSlot(Record r) throws IOException {
        ByteBuffer pageData = bufferManager.getPageData(pid);
        BitSet slotMap = readSlotMap(pageData);
        slotMap.clear(r.getSid());
        pageData.position(SLOT_MAP_POSITION);
        pageData.put(slotMap.toByteArray());
        pageData.position(0);
    }

    public Object getRecordColumnValue(int sid, Column column) throws IOException {
        int size = getSlotSize();
        int dataIndex = getDataIndex();
        int position = dataIndex + size * sid + column.getOffset();
        Object res = column.getType().readValueFromByteBuffer(bufferManager.getPageData(pid), position); 
        return res;
    }

    public byte[] toByteArray(int sid) throws IOException {
        ByteBuffer data = bufferManager.getPageData(pid);
        byte[] dst = new byte[getSlotSize()];
        int position = data.position();
        data.position(getDataIndex() + getSlotSize() * sid);
        data.get(dst, 0, dst.length);
        data.position(position);
        return dst;
    }



    private void writeSlotMap(BitSet slotMap) throws IOException {
        ByteBuffer pageData = bufferManager.getPageData(pid);
        pageData.position(SLOT_MAP_POSITION);
        pageData.put(slotMap.toByteArray());
    }

    private int getDataIndex() throws IOException {
        return bufferManager.getPageData(pid).getInt(DATA_INDEX_POSITION);
    }

    private BitSet readSlotMap(ByteBuffer pageData) throws IOException {
        int slotMapSize = pageData.getInt(SLOT_MAP_SIZE_POSITION);
        int oldLimit = pageData.limit();
        pageData.position(SLOT_MAP_POSITION);
        pageData.limit(SLOT_MAP_POSITION + slotMapSize);
        BitSet slotMap = BitSet.valueOf(pageData);
        pageData.limit(oldLimit);
        return slotMap;
    }

    private boolean checkAllConditions(Record r, List<Condition> conditions) throws IOException {
        for(Condition c : conditions) {
            if(!c.check(r)) {
                return false;
            }
        }
        return true;
    }
    private final int pid;

    private final static int RECORD_SIZE_POSITION = 0;
    private final static int SLOT_MAP_SIZE_POSITION = Integer.SIZE / 8;
    private final static int DATA_INDEX_POSITION = 2 * Integer.SIZE / 8;
    private final static int SLOT_MAP_POSITION = 3 * Integer.SIZE / 8;

    private final static BufferManager bufferManager = BufferManager.getInstance();

//
//    public Collection<Record> deleteRecords(Condition c) throws IOException {
//        ArrayList<Record> records = new ArrayList<Record>();
//        BitSet tmpSlotMap = (BitSet) slotMap.clone();
//        int i = tmpSlotMap.nextSetBit(0);
//        while(i >= 0) {
//            Record r = new Record(pid, i, slotSize);
//            if(c.check(r)) {
//                records.add(r);
//                tmpSlotMap.clear(i*slotSize,(i+1)*slotSize);
//            }
//        }
//        slotMap = tmpSlotMap;
//        return records;
//    }
//
//
//    public Collection<Record> bulkInsertRecord(ByteBuffer[] recordData) throws IOException {
//        BitSet tmpSlotMap = (BitSet) slotMap.clone();
//        Collection<Record> records = new ArrayList<Record>();
//        for (ByteBuffer data : recordData){
//             int i = tmpSlotMap.nextClearBit(0);
//            tmpSlotMap.set(i);
//            Record newRecord =  new Record(pid, i, slotSize);
//            newRecord.setInfo(data);
//            records.add(newRecord);
//        }
//        slotMap = tmpSlotMap;
//        return records;
//    }



}
